package com.wyt.common.sqljdbc;

import com.alibaba.fastjson.JSONObject;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.wyt.common.sqljdbc.bean.ColumnInfo;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.text.SimpleDateFormat;
import java.util.*;


@Component
public class MysqlBinlogListener implements ApplicationRunner {
    private static Logger logger = LoggerFactory.getLogger(MysqlBinlogListener.class);

    // database:{table:host}
    private static Map<String, Map<String, String>> dbTableHostMap = new HashMap<>();
    // tableId:(table:host)
    private static Map<Long, String> tableIdNameMap = new HashMap<>();

    private static Map<Long, List<ColumnInfo>> tableIdColumns = new HashMap<>();

    @Autowired
    private RestTemplate restTemplate;

    @Value("${mysql.binlog.status}")
    private Boolean binlog = true;
    @Value("${mysql.binlog.syncHost}")
    private String syncHost = "127.0.0.1";
    @Value("${mysql.binlog.host}")
    private String listenMysqlHost = "127.0.0.1";
    @Value("${mysql.binlog.port}")
    private Integer listenMysqlPort = 3306;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("**************** 开始初始化binlog监听工具 *********************");
        if (binlog){
            //获取待同步的表信息
            getDbTableHostMap();
            //启动线程，监听binlog
            new Thread(() -> {
                try {
                    listen();
                } catch (Exception e) {
                    logger.error("binlog监听工具异常退出，请排查原因", e);
                }
            }).start();
        }

        logger.info("**************** binlog监听初始化完成 *********************\n");
    }


    public Map<String, Map<String, String>> getDbTableHostMap() {
        Map<String, String> tableHostMap = new HashMap<>();
        tableHostMap.put("zzll_bd_wqzb",syncHost);
        tableHostMap.put("zzll_bd_wqzb_dtqk", syncHost);
        tableHostMap.put("task_order", syncHost);
        tableHostMap.put("task_link_node", syncHost);
        dbTableHostMap.put("datamf", tableHostMap);
        logger.info("要同步的表信息有{}", JSONObject.toJSONString(dbTableHostMap));
        return dbTableHostMap;
    }


    public void listen() throws Exception {
        BinaryLogClient client = new BinaryLogClient(listenMysqlHost, listenMysqlPort, "root", "123456");
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG);
        client.setEventDeserializer(eventDeserializer);
        client.setServerId(1);

        client.registerEventListener(event -> {
            EventData data = event.getData();
            EventType eventType = event.getHeader().getEventType();
            switch (eventType) {
                case TABLE_MAP:
                    tableMapEventHandler(data);
                    break;
                case EXT_WRITE_ROWS:
                    writeRowEventHandler(data);
                    break;
                case EXT_DELETE_ROWS:
                    deleteRowEventHandler(data);
                    break;
                case EXT_UPDATE_ROWS:
                    updateRowEventHandler(data);
                    break;
                default:
                    logger.warn("暂不支持{}类型的事件处理", eventType);
            }
        });
        client.connect();
    }

    public void tableMapEventHandler(EventData data) {
        TableMapEventData tData = (TableMapEventData) data;
        String database = tData.getDatabase();
        String table = tData.getTable();
        long tableId = tData.getTableId();
        logger.info("database:{},tableName:{},tableId:{}", database, table, tableId);
        if (dbTableHostMap.containsKey(database) && dbTableHostMap.get(database).containsKey(table)) {
            if (tableIdNameMap.containsKey(tableId)) {
                logger.info("已经记录tableId:{}对应的同步信息{}，不进行重复记录", tableId, tableIdNameMap.get(tableId));
            } else {
                logger.info("{}.{}是要同步的数据，记录其tableId:{}信息", database, table, tableId);
                tableIdNameMap.put(tableId, database + ":" + table + ":" + dbTableHostMap.get(database).get(table));
            }
        }
    }

    public void writeRowEventHandler(EventData data) {
        WriteRowsEventData wData = (WriteRowsEventData) data;
        long tableId = wData.getTableId();
        if (!tableIdNameMap.containsKey(tableId)) return;
        String tableNameHost = tableIdNameMap.getOrDefault(tableId, "::");
        String database = tableNameHost.split(":")[0];
        String tableName = tableNameHost.split(":")[1];
        String host = tableNameHost.split(":")[2];
        logger.info("table:{} {}接收到insert事件，向{}主机同步", tableName, tableId, host);
        BitSet includedColumns = wData.getIncludedColumns();
        List<Serializable[]> rows = wData.getRows();
        Object[] rowValues = Arrays.stream(rows.get(0)).toArray();
        StringBuffer sb = new StringBuffer();
        sb.append("insert into ").append(tableName).append(" values(");
        for (int i = 0; i < includedColumns.length(); i++) {
            if (rowValues[i] == null) {
                sb.append("null");
            } else {
                if (rowValues[i] instanceof Number) {
                    sb.append(rowValues[i]);
                } else {
                    sb.append("'").append(rowValues[i]).append("'");
                }
            }
            sb.append(",");
        }
        sb.deleteCharAt(sb.length() - 1);
        sb.append(")");
        String sql = sb.toString();
        logger.info("生成insert sql:{}", sql);
        sendSyncReq(sql, host);
    }


    public void deleteRowEventHandler(EventData data) {
        DeleteRowsEventData dData = (DeleteRowsEventData) data;
        long tableId = dData.getTableId();
        if (!tableIdNameMap.containsKey(tableId)) return;
        String tableNameHost = tableIdNameMap.getOrDefault(tableId, "::");
        String database = tableNameHost.split(":")[0];
        String tableName = tableNameHost.split(":")[1];
        String host = tableNameHost.split(":")[2];
        if (!tableIdColumns.containsKey(tableId)) {
            tableIdColumns.put(tableId, getColumns(database, tableName));
        }
        List<ColumnInfo> columns = tableIdColumns.get(tableId);
        logger.info("table:{} {}接收到delete事件，向{}主机同步", tableName, tableId, host);
        List<Serializable[]> rows = dData.getRows();
        StringBuffer sb = new StringBuffer();
        for (Serializable[] row : rows) {
            sb.append("delete from ").append(tableName).append(" where ");
            Object[] rowValues = Arrays.stream(row).toArray();
            for (int i = 0; i < columns.size(); i++) {
                if ("datetime".equalsIgnoreCase(columns.get(i).getDataType())) continue;
                if (rowValues[i] == null) continue;
                sb.append(columns.get(i).getColName()).append("=");
                if (rowValues[i] instanceof Number) {
                    sb.append(rowValues[i]);
                } else {
                    sb.append("'").append(rowValues[i]).append("'");
                }
                sb.append(" and ");
            }
            sb.delete(sb.length() - 4, sb.length());
        }
        logger.info("生成delete sql:{}", sb);
        sendSyncReq(sb.toString(), host);
    }

    public void updateRowEventHandler(EventData data) {
        UpdateRowsEventData udata = (UpdateRowsEventData) data;
        long tableId = udata.getTableId();
        if (!tableIdNameMap.containsKey(tableId)) return;
        String tableNameHost = tableIdNameMap.getOrDefault(tableId, "::");
        String database = tableNameHost.split(":")[0];
        String tableName = tableNameHost.split(":")[1];
        String host = tableNameHost.split(":")[2];
        if (!tableIdColumns.containsKey(tableId)) {
            tableIdColumns.put(tableId, getColumns(database, tableName));
        }
        List<ColumnInfo> columns = tableIdColumns.get(tableId);
        logger.info("table:{} {}接收到update事件，向{}主机同步", tableName, tableId, host);
        List<Map.Entry<Serializable[], Serializable[]>> rows = udata.getRows();
        StringBuffer sb = new StringBuffer();
        for (Map.Entry<Serializable[], Serializable[]> row : rows) {
            List<Serializable> entries = Arrays.asList(row.getValue());
            Object[] rowValues = entries.toArray();
            sb.append("update ").append(tableName).append(" set ");
            for (int i = 0; i < columns.size(); i++) {
                sb.append(columns.get(i).getColName()).append("=");
                if (rowValues[i] == null) {
                    sb.append("null");
                } else {
                    if (rowValues[i] instanceof Number) {
                        if ("datetime".equalsIgnoreCase(columns.get(i).getDataType())) {
                            sb.append("'").append(getDateStr((long) rowValues[i])).append("'");
                        } else {
                            sb.append(rowValues[i]);
                        }
                    } else {
                        sb.append("'").append(rowValues[i]).append("'");
                    }
                }
                sb.append(",");
            }
            sb.deleteCharAt(sb.length() - 1);
            sb.append(" where ");
            Object[] oldValues = Arrays.stream(row.getKey()).toArray();
            for (int i = 0; i < columns.size(); i++) {
                //时间类型字段不能作为更新的key
                if ("datetime".equalsIgnoreCase(columns.get(i).getDataType())) continue;
                //变更的字段不能作为更新的key
                if (oldValues[i] != rowValues[i]) continue;
                sb.append(columns.get(i).getColName()).append("=");
                if (oldValues[i] == null) {
                    sb.append("null");
                } else {
                    if (oldValues[i] instanceof Number) {
                        sb.append(oldValues[i]);
                    } else {
                        sb.append("'").append(oldValues[i]).append("'");
                    }
                }
                sb.append(" and ");
            }
            sb.delete(sb.length() - 4, sb.length());
        }
        logger.info("生成update sql:{}", sb);
        sendSyncReq(sb.toString(), host);
    }

    public static String getDateStr(long time) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.getDefault());
        String dateString = sdf.format(new Date(time));
        return dateString;
    }

    public List<ColumnInfo> getColumns(String db, String table) {
        List<ColumnInfo> columns = new ArrayList<>();
        try {
            logger.info("开始获取{}.{}的字段信息", db, table);
            Class.forName("com.mysql.jdbc.Driver");
            // 保存当前注册的表的column信息
            Connection connection = DriverManager.getConnection("jdbc:mysql://"+listenMysqlHost+":"+listenMysqlPort+"/", "root", "123456");
            // 执行sql
            String preSql = "SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, " +
                    "DATA_TYPE, ORDINAL_POSITION, case when COLUMN_KEY = 'PRI' THEN 'Y' ELSE 'N' END IS_PKC" +
                    " FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = ? and TABLE_NAME = ? order by ORDINAL_POSITION";
            PreparedStatement ps = connection.prepareStatement(preSql);
            ps.setString(1, db);
            ps.setString(2, table);
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                String schema = rs.getString("TABLE_SCHEMA");
                String tableName = rs.getString("TABLE_NAME");
                String column = rs.getString("COLUMN_NAME");
                int idx = rs.getInt("ORDINAL_POSITION");
                String dataType = rs.getString("DATA_TYPE");
                String isPKC = rs.getString("IS_PKC");
                ColumnInfo info = new ColumnInfo(schema, tableName, idx - 1, column, dataType, "Y".equals(isPKC));
                columns.add(info);
            }
            ps.close();
            rs.close();
        } catch (Exception e) {
            logger.error("load db conf error, db_table={}:{} ", db, table, e);
        }
        logger.info("获取到{}.{}的字段信息是{}", db, table, JSONObject.toJSONString(columns));
        return columns;
    }

    private void sendSyncReq(String sql, String host) {
        if (StringUtils.isBlank(sql)) return;

        JSONObject request = new JSONObject();
        String url = "http://" + host + ":32624/dds/receiveSyncSql";
        String s = restTemplate.postForObject(url, request, String.class);

        logger.info("向主机{}发送同步sql：{}，响应结果为{}", host, sql, s);
    }


    public static void main(String[] args)throws Exception {
        MysqlBinlogListener listener = new MysqlBinlogListener();
        listener.run(null);
    }

}
